package com.lsx143.realtime.function;

import com.lsx143.realtime.util.JdbcUtil;
import com.lsx143.realtime.util.RedisUtil;
import com.lsx143.realtime.util.ThreadPoolUtil;
import lombok.SneakyThrows;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.util.concurrent.ThreadPoolExecutor;

public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> {

    private ThreadPoolExecutor threadPool;

    protected abstract void addDim(Connection phoenixCon, Jedis jedis, T t, ResultFuture<T> resultFuture);

    /**
     * 获取一个连接池
     *
     * @param parameters 配置参数
     */
    @Override
    public void open(Configuration parameters) {
        threadPool = ThreadPoolUtil.getThreadPool();
    }

    /*@Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        System.err.println("time out ..." + input);
//        resultFuture.complete();
    }*/

    /**
     * 异步调用这个方法
     *
     * @param t            传入维度层的泛型
     * @param resultFuture 容器
     */
    @Override
    public void asyncInvoke(T t, ResultFuture<T> resultFuture) {
        threadPool.submit(new Runnable() {
            @SneakyThrows
            @Override
            public void run() {
                Connection phoenixCon = null;
                Jedis jedis = null;
                try {
                    jedis = RedisUtil.getRedisClient();
                    phoenixCon = JdbcUtil.getPhoenixConnection();
                    addDim(phoenixCon, jedis, t, resultFuture);
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    if (phoenixCon != null) {
                        phoenixCon.close();
                    }
                    if (jedis != null) {
                        jedis.close();
                    }
                }
            }
        });
    }
}
